package com.yl.test;

import cn.hutool.core.date.DateUtil;
import com.yl.entity.MultiDataEntity;
import com.yl.util.JdkUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.yarn.webapp.hamlet.Hamlet;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

/**
 * @author wlf
 * @since 2022/8/9
 */
@Slf4j
public class WindowExample {

    public static void main(String[] args) throws Exception {
        JdkUtil.disableWarning();
        // 流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // socket 数据源
        DataStreamSource<String> streamSource = env.socketTextStream("elephant", 6666);

        WatermarkStrategy<String> watermarkStrategy = WatermarkStrategy
                .forGenerator(new WatermarkGeneratorSupplier<String>() {
                    @Override
                    public WatermarkGenerator<String> createWatermarkGenerator(Context context) {
                        return new WatermarkGenerator<String>() {
                            private final long delay = 0;
                            private long maxTs;
                            @Override
                            public void onEvent(String s, long eventTs, WatermarkOutput watermarkOutput) {
                                log.info("maxTs: {} eventTs: {}", maxTs, eventTs);
                                maxTs = Math.max(maxTs, eventTs);
                            }
                            @Override
                            public void onPeriodicEmit(WatermarkOutput output) {
//                                log.info("emitWatermark: {}", maxTs - delay);
                                output.emitWatermark(new Watermark(maxTs - delay));
                            }
                        };
                    }
                })
//                .<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((s, timestamp) -> {
                    long ts = Long.parseLong(s.split(",")[0]);
                    log.info("ts:{}", DateUtil.date(ts).toString());
                    return DateUtil.date(ts).getTime();
                });

        streamSource
                .assignTimestampsAndWatermarks(watermarkStrategy)
                .map(new MapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String s) throws Exception {
                        String[] split = s.split(",");
                        return Tuple2.of(split[1], 1);
                    }
                })
                .keyBy((KeySelector<Tuple2<String, Integer>, String>) ele -> ele.f0)
                .window(TumblingEventTimeWindows.of(Time.seconds(1)))
                .process(new ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String, TimeWindow>.Context context, Iterable<Tuple2<String, Integer>> elements, Collector<Tuple2<String, Integer>> out) throws Exception {
                        String start = DateUtil.date(context.window().getStart()).toString();
                        String end = DateUtil.date(context.window().getEnd()).toString();
                        String watermark = DateUtil.date(context.currentWatermark()).toString();
                        log.info("start:{} end:{} watermark:{}", start, end, watermark);
                        Map<String, Integer> kvs = new HashMap<>();
                        for (Tuple2<String, Integer> element : elements) {
                            if (kvs.containsKey(element.f0)) {
                                kvs.put(element.f0, kvs.get(element.f0) + 1);
                            } else {
                                kvs.put(element.f0, 1);
                            }
                        }
                        for (String key : kvs.keySet()) {
                            out.collect(Tuple2.of(key, kvs.get(key)));
                        }
                    }
                })
                .print();

        System.out.println("### 流配置完毕 ###");
        // 懒加载
        env.execute();
    }
}
